/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package com.vasilrem.linked

import scala.actors._
import scala.actors.Actor._
import scala.collection.immutable.Queue
import java.util.Date

object ChildRestartBehavior extends Enumeration{
  type ChildRestartBehavior = Value
  val permanent, transient, temporary = Value
}
import ChildRestartBehavior._
case class ChildSpecification(worker: Worker, restartBehavior: ChildRestartBehavior.Value = permanent)

sealed trait RestartStrategy{
  val maxRestarts: Long
  val maxTime: Long
}
case class OneForOne(maxRestarts: Long = 3, maxTime: Long = 3000) extends RestartStrategy
case class OneForAll(maxRestarts: Long = 3, maxTime: Long = 3000) extends RestartStrategy

trait Worker extends Actor{
  val id: String

  private[linked] var restarts = List[Date]()

  override def restart = synchronized {
    restarts = new Date :: restarts
    super.restart()
  }

  protected[linked] def restartsInPeriod(period: Long):Int = synchronized {
    val threshold = new Date(new Date().getTime - period)
    restarts.filter(_.after(threshold)).size
  }

  protected[linked] lazy val getExitReason = {
    val linksField = getClass.getDeclaredField("exitReason")
    linksField.setAccessible(true)
    linksField.get(this).asInstanceOf[AnyRef]
  }
}

case class Terminate(reason: AnyRef)

class Supervisor(supervisorId: String, strategy: RestartStrategy, childSpecs: List[ChildSpecification]) extends Worker {

  val id: String = supervisorId

  private[linked] def linkWorkers = synchronized {
    childSpecs.foreach{spec =>
      println("Starting worker %s" format(spec.worker.id))
      self.link(spec.worker.getState match {
          case State.New => spec.worker.start()
          case State.Terminated => spec.worker.restart(); spec.worker
          case _ => spec.worker
        })
    }
  }
  
  def getSpec(id: String) = childSpecs.find{spec => spec.worker.id == id}

  private[linked] def restartWorker(worker: Worker) = synchronized {
    for{
      spec <- childSpecs.find(_.worker.id == worker.id)
    } if(spec.restartBehavior == permanent ||
         (spec.restartBehavior == transient && worker.getExitReason == 'normal)){
      println("Restarting worker [%s]..." format(worker.id))
      if(spec.worker.getState == State.Terminated)
        spec.worker.restart
      self link worker
    } else println("%s worker [%s] will not be restarted." format(spec.restartBehavior, spec.worker.id))
  }
  
  override def act = {
    self.trapExit = true
    linkWorkers
    loop {
      react{
        case Exit(worker: Worker, reason) =>
          println("Worker [%s] has failed due to [%s]" format(worker.id, reason))
          if(worker.restartsInPeriod(strategy.maxTime) >= strategy.maxRestarts) exit("Maximum restart intensity for %s is reached!" format(worker.id))
          strategy match {
            case str:OneForOne => restartWorker(worker)
            case str:OneForAll => childSpecs.foreach{spec => restartWorker(spec.worker)}
          }
        case Terminate(reason) => println("Supervisor terminated with reason [%s]" format(reason))
          exit(reason)
      }
    }
  }

}

object Supervisor {

  def apply(supervisorId: String, strategy: RestartStrategy = OneForOne(), childSpecs: List[ChildSpecification] = List[ChildSpecification]()) =
    new Supervisor(supervisorId, strategy, childSpecs)

}

